package com.hugeyurt.headers;
import com.rabbitmq.client.*;

import java.util.HashMap;
import java.util.Map;

public class ConsumerHeaders
{
    private final static String QUEUE_NAME = "my_queue_headers";
    private final static String exchange_name = "my_exchange_headers2";

    public static void main(String[] args) throws Exception
    {

        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost"); // RabbitMQ服务器地址
        factory.setUsername("guest"); // 默认用户名
        factory.setPassword("guest"); // 默认密码
        factory.setPort(5672);//设置端口号
        // 声明队列（需要与生产者配置一致）
        //创建连接对象
        Connection connection = factory.newConnection();
        //创建信道
        Channel   channel = connection.createChannel();

        // 1. 声明交换机
        channel.exchangeDeclare(exchange_name, BuiltinExchangeType.HEADERS,false,false,null );
        // 2. 创建队列（临时队列）
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 3. 定义绑定规则（Headers匹配条件）
        Map<String, Object> headers = new HashMap<>();
        headers.put("x-match", "all");  // 匹配所有条件（可改为"any"匹配任意条件）
        headers.put("key1", "hugeyurt22");
        headers.put("key2", "hugeyurt22");
        headers.put("format", "pdf");
        // 4. 绑定队列到交换机
       channel.queueBind(QUEUE_NAME, exchange_name, "", headers);


        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            Map<String, Object> receivedHeaders = delivery.getProperties().getHeaders();
            System.out.println(" [x] Received: '" + message + "'\nHeaders: " + receivedHeaders);
            // 手动消息确认（需要关闭自动确认）
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

        };

        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println(" [x] Cancelled");
        };

        // 消费消息（关闭自动确认）
        /*
         1 消费哪个队列，指定队列名称
         2 消费成功后，是否自动应答；true 自动应答
         3、接收消息的回调函数
         4、取消消息的回调函数

         */
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, cancelCallback);
    }
}